package cn.xinfei.xdecision.pulsar;

import cn.xinfei.xdecision.common.utils.util.IpUtil;
import cn.xinfei.xdecision.pulsar.config.XPulsarConsumerProperties;
import cn.xinfei.xdecision.pulsar.config.XPulsarProducerProperties;
import cn.xinfei.xdecision.pulsar.config.XPulsarProperties;
import cn.xinfei.xdecision.pulsar.consumer.XPulsarConsumer;
import cn.xinfei.xdecision.pulsar.model.XPulsarConsumerModel;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.math.RandomUtils;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.impl.schema.StringSchema;
import org.python.netty.util.concurrent.DefaultThreadFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
@ConditionalOnProperty(name = "spring.xpulsar.enable")
@Configuration
@ConditionalOnBean(value = XPulsarProperties.class)
@Order(1)
public class XPulsarClientFactory implements ApplicationListener<ApplicationReadyEvent>, ApplicationContextAware {

    @Autowired
    private XPulsarProperties xPulsarProperties;

    private static volatile PulsarClient pulsarClient;
    private static final ConcurrentHashMap<String, Producer<String>> producer = new ConcurrentHashMap<>();
    private static final ConcurrentHashMap<String, Consumer<String>> consumer = new ConcurrentHashMap<>();
    private static final ConcurrentHashMap<String, XPulsarConsumerProperties> consumerPropertiesConcurrentHashMap = new ConcurrentHashMap<>();
    private static final ConcurrentHashMap<String, XPulsarProducerProperties> producerPropertiesConcurrentHashMap = new ConcurrentHashMap<>();
    private static ApplicationContext context;

    private static XPulsarProducerProperties producerProperties;
    private static XPulsarConsumerProperties consumerProperties;

    private static Object producerLock = new Object();
    public static Object consumerLock = new Object();


    @Override
    public void onApplicationEvent(ApplicationReadyEvent applicationStartedEvent) {
        log.info("==========ApplicationReadyEvent===ProducerClientFactory================");
        initWithRetry();
//        executorService.scheduleWithFixedDelay(this::initWithRetry, 0, 1, TimeUnit.SECONDS);
    }

    private void initWithRetry() {
        try {
            XPulsarClientFactory.pulsarClient = PulsarClient.builder()
                    .serviceUrl(xPulsarProperties.getServiceUrl())
//                    .ioThreads(xPulsarProperties.getIoThreads())
//                    .listenerThreads(xPulsarProperties.getListenerThreads())
//                    .operationTimeout(xPulsarProperties.getOperationTimeoutSec(), TimeUnit.SECONDS)
//                    .connectionTimeout(xPulsarProperties.getConnectionTimeoutSec(), TimeUnit.SECONDS)
//                    .connectionsPerBroker(xPulsarProperties.getConnectionsPerBroker())
                    .build();
            log.info("pulsar client init success");
            consumerProperties = xPulsarProperties.getConsumer();
            cacheConsumerProperties(consumerProperties);

            producerProperties = xPulsarProperties.getProducer();
            cacheProducerProperties(producerProperties);

//            this.executorService.shutdown();
        } catch (Exception e) {
            log.error("init pulsar error, exception is ", e);
        }
    }

    public static Producer<String> getProducer(String topic) {
        Producer<String> producer = XPulsarClientFactory.producer.get(topic);
        if(producer==null){
            log.info("producer is null. topic = {}",topic);
            newProducer(producerProperties, topic);
            producer = XPulsarClientFactory.producer.get(topic);
        }
        return producer;
    }


    private synchronized static void newProducer(XPulsarProducerProperties producerProperties, String topic) {
        try {
            String topicName = "persistent://" + producerProperties.getTenant() + "/" + producerProperties.getNamespace() + "/" + topic;

            Producer<String> stringProducer = null;
            if (producerProperties.isUseDefault()) {
                stringProducer = pulsarClient.newProducer(Schema.STRING)
                        .topic(topicName).create();
                log.info("pulsar producer created success by default .topic={}", topicName);
            } else {
                stringProducer = pulsarClient.newProducer(Schema.STRING)
                        .topic(topicName)
                        .batchingMaxPublishDelay(producerProperties.getBatchingMaxPublishDelay(), TimeUnit.MILLISECONDS)
                        .sendTimeout(producerProperties.getSendTimeout(), TimeUnit.SECONDS)
                        .blockIfQueueFull(producerProperties.isBlockIfQueueFull())
                        .batchingMaxMessages(producerProperties.getBatchingMaxMessages())
                        .batchingMaxBytes(producerProperties.getBatchingMaxBytes())
                        .enableBatching(producerProperties.isEnableBatching())
                        .maxPendingMessages(producerProperties.getMaxPendingMessages())
//                        .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
                        .create();
                log.info("pulsar producer created success by customer .topic={}", topicName);
            }
            log.info("pulsar producer created success.topicName={},topic={}", topicName,topic);
            producer.put(topic, stringProducer);
        } catch (PulsarClientException e) {
            log.error("pulsar producer fail.topic={},error={}", topic, e.getMessage(), e);
        }
    }


    public synchronized static XPulsarConsumer<String> newConsumer(XPulsarConsumerModel xPulsarConsumerModel) {
        XPulsarConsumerProperties consumerProperties = null;
        String topic = xPulsarConsumerModel.getTopic();
        String subscriptionName = xPulsarConsumerModel.getSubscriptionName();
        String consumerName = xPulsarConsumerModel.getConsumerName();
        try {
            consumerProperties = consumerPropertiesConcurrentHashMap.get(topic);
            if (consumerProperties == null) {
                log.error("配置信息不存在，topic={}", topic);
                return null;
            }
            String t = "persistent://" + consumerProperties.getTenant() + "/" + consumerProperties.getNamespace() + "/" + topic;
            log.info("subscribe topic={},subscriptionName={},consumerName={}", t, subscriptionName, consumerName);
            Consumer<String> consumer = pulsarClient.newConsumer(new StringSchema())
                    .topic(t)
//                    .consumerName(consumerName+"_"+IpUtil.getIpAddress()+"_"+RandomUtils.nextInt(1000))
                    .subscriptionName(subscriptionName)
                    .subscriptionMode(SubscriptionMode.valueOf(consumerProperties.getSubscriptionMode()))
                    .subscriptionType(SubscriptionType.valueOf(consumerProperties.getSubscriptionType()))
//                    .ackTimeout(consumerProperties.getAckTimeout(), TimeUnit.MILLISECONDS)
//                    .ackTimeoutTickTime(consumerProperties.getAckTimeoutTickTime(), TimeUnit.MILLISECONDS)
//                    .enableRetry(consumerProperties.isEnableRetry())
//                    .enableBatchIndexAcknowledgment(consumerProperties.isEnableBatchIndexAcknowledgment())
//                    .isAckReceiptEnabled(consumerProperties.isAckReceiptEnabled())
//                    .poolMessages(true)
                    .autoUpdatePartitions(true)
                    .subscribe();
            XPulsarConsumer xPulsarConsumer = new XPulsarConsumer(consumer, xPulsarConsumerModel);
            log.info("pulsar consumer created success.{}={}", topic, consumer.getConsumerName());
            Thread.currentThread().setName(topic + "_" + xPulsarConsumerModel.getConcurrentNum() + "_" + consumerName);
            return xPulsarConsumer;
        } catch (Exception e) {
            log.error("getConsumer fail.topic={},subscriptionName={},consumerName={}.consumerProperties={},error={}", topic, subscriptionName, consumerName, JSONObject.toJSONString(consumerProperties), e.getMessage(), e);
        }
        return null;

    }


    private void cacheConsumerProperties(XPulsarConsumerProperties consumerProperties) {
        if (consumerProperties == null) {
            log.info("pulsar consumer not need init");
            return;
        }
        List<String> topics = consumerProperties.getTopics();
        if (CollectionUtils.isEmpty(topics) == false) {
            for (int i = 0; i < topics.size(); i++) {
                String topic = topics.get(i);
                log.info("pulsar consumer init properties.topic={}", topic);
                consumerPropertiesConcurrentHashMap.put(topic, consumerProperties);
            }
        }
    }

    /**
     * toto 增加重试机制
     *
     * @param producerProperties
     */
    private void cacheProducerProperties(XPulsarProducerProperties producerProperties) {
        if (producerProperties == null) {
            log.info("pulsar producer not need init");
            return;
        }
        List<String> topics = producerProperties.getTopics();
        if (CollectionUtils.isEmpty(topics) == false) {
            for (int i = 0; i < topics.size(); i++) {
                String topic = topics.get(i);
                log.info("pulsar producer init properties.topic={}", topic);
                producerPropertiesConcurrentHashMap.put(topic, producerProperties);
                log.info("create producer topic={}", topic);
                newProducer(producerProperties, topic);
            }
        }
    }


    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        context = applicationContext;
    }
}
